package com.manlitech.cloud.stream.controller;

import com.manlitech.cloud.stream.componet.rocketmq.Producer;
import com.manlitech.cloud.stream.config.rocketmq.RmqConfig;
import com.manlitech.cloud.stream.pojo.ProductOrder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;

/**
 * 普通rocketmq发送消息
 *  同步
 *  异步
 *  一次性
 *  延迟
 *  顺序队列
 *  事务
 * @author shensg
 * @date 2021/7/29 15:48
 */
@Slf4j
@RestController
public class TestNomalRocketMqController {

    @Autowired
    private Producer producer;

    @Autowired
    RmqConfig rmqConfig;

    @RequestMapping("/rocketmq/send")
    public Object callback() throws Exception {
        for (int i = 0; i < 100; i++) {
            // 创建生产信息
            Message message = new Message(rmqConfig.TOPIC, "testTag", (String.format("[生产者] 发送第 【%d】 次 mq 消息", i)).getBytes());
            // 发送消息
            SendResult sendResult = producer.getProducer().send(message);
            log.info("  >>> [生产者] 发送消息返回对象 :: {}", sendResult);
        }

        return "执行成功";
    }

    /**
     * 发送普通消息（三种方式)
     */

    /**
     * 1、同步发送消息
     */
    @RequestMapping("/rocketmq/sync")
    private  void sync() throws Exception {
        //创建消息
        Message message = new Message(rmqConfig.TOPIC, ("  同步发送  ").getBytes());
        //同步发送消息
        SendResult sendResult = producer.getProducer().send(message);
        log.info("Product-同步发送-Product信息={}", sendResult);
    }

    /**
     * 2、异步发送消息
     */
    @RequestMapping("/rocketmq/async")
    private  void async() throws Exception {
        //创建消息
        Message message = new Message(rmqConfig.TOPIC, ("  异步发送  ").getBytes());
        //异步发送消息
        producer.getProducer().send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("Product-异步发送-输出信息={}", sendResult);
            }
            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                //补偿机制，根据业务情况进行使用，看是否进行重试
            }
        });
    }

    /**
     * 3、单项发送消息
     */
    @RequestMapping("/rocketmq/oneWay")
    private  void oneWay() throws Exception {
        //创建消息
        Message message = new Message(rmqConfig.TOPIC, (" 单项发送 ").getBytes());
        //同步发送消息
        producer.getProducer().sendOneway(message);
    }

    /**
     * 4、发送延迟消息
     */
    @RequestMapping("/rocketmq/delay")
    private  void delay() throws Exception {
        //创建消息
        Message message = new Message(rmqConfig.TOPIC, (" 延迟发送 ").getBytes());
        // messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        // 这里设置4，即30s的延迟
        message.setDelayTimeLevel(2);
        //延迟发送消息
        SendResult sendResult = producer.getProducer().send(message);
        log.info("Product-延迟发送-Product信息={}", sendResult);
    }

    /**
     * RocketMQ顺序消费
     * 1、顺序消息暂不支持广播模式。
     * 2、顺序消息不支持异步发送方式，否则将无法严格保证顺序。
     * 3、建议同一个 Group ID 只对应一种类型的 Topic，即不同时用于顺序消息和无序消息的收发。
     * 4、对于全局顺序消息，建议创建实例个数 >=2。
     */
    private static List<ProductOrder> orderList = null;
    private static String producerGroup = "test_producer";
    /**
     * 模拟数据
     */
    static {
        orderList = new ArrayList<>();
        orderList.add(new ProductOrder("XXX001", "订单创建"));
        orderList.add(new ProductOrder("XXX001", "订单付款"));
        orderList.add(new ProductOrder("XXX001", "订单完成"));
        orderList.add(new ProductOrder("XXX002", "订单创建"));
        orderList.add(new ProductOrder("XXX002", "订单付款"));
        orderList.add(new ProductOrder("XXX002", "订单完成"));
        orderList.add(new ProductOrder("XXX003", "订单创建"));
        orderList.add(new ProductOrder("XXX003", "订单付款"));
        orderList.add(new ProductOrder("XXX003", "订单完成"));
    }

    @RequestMapping("/rocketmq/listmessage")
    public  void sendMessage() throws Exception {
        for (ProductOrder order : orderList) {
            //1、生成消息
            Message message = new Message(rmqConfig.TOPIC, "", order.getOrderId(), order.toString().getBytes());
            //2、发送消息是 针对每条消息选择对应的队列
            SendResult sendResult = producer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    //3、arg的值其实就是下面传入 orderId
                    String orderid = (String) arg;
                    //4、因为订单是String类型，所以通过hashCode转成int类型
                    int hashCode = orderid.hashCode();
                    //5、因为hashCode可能为负数 所以取绝对值
                    hashCode = Math.abs(hashCode);
                    //6、保证同一个订单号 一定分配在同一个queue上
                    long index = hashCode % mqs.size();
                    return mqs.get((int) index);
                }
            }, order.getOrderId(),50000);

            System.out.printf("Product：发送状态=%s, 存储queue=%s ,orderid=%s, type=%s\n", sendResult.getSendStatus(),
                    sendResult.getMessageQueue().getQueueId(), order.getOrderId(), order.getType());
        }
    }

}
